package com.suxin.socket.vaca.service;

import com.alibaba.fastjson.JSON;
import com.suxin.socket.vaca.config.WebSocketServer;
import com.suxin.socket.vaca.model.MqMessage;
import com.suxin.socket.vaca.model.User;
import com.suxin.socket.vaca.model.WsResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

/**
 * @author Tang
 * @classname BroadCastMessage
 * @description [  ]
 * @date 2021/4/1 17:12
 */
@Slf4j
public class BroadCastMessage {

    private static final String TOPIC = "sx_vaca_cc";

    @Resource
    private KafkaTemplate<Integer, String> kafkaTemplate;

    /**
     * 广播消息
     * @param mqMessage
     */
    private void broadcast(MqMessage mqMessage) {
        kafkaTemplate.send(TOPIC, mqMessage.toString());
    }

    @KafkaListener(topics = {TOPIC})
    private void receive(ConsumerRecord<String,String> record) {
        MqMessage mqMessage = JSON.parseObject(record.value(), MqMessage.class);
        this.wsPush(mqMessage);
    }

    /**
     * websocket 长连接推送
     * @param mqMessage
     */
    private void wsPush(MqMessage mqMessage) {
        mqMessage.getUserIds().forEach((item) -> {
            User user = WebSocketServer.CLIENTS.get(item);
            if (user != null) {
                WsResponse wsResponse = mqMessage.getResponse();
                user.send(wsResponse.toString());
                log.info("向客户端用户[{}]推送消息:{}",user.getId() + "-"+user.getName(), wsResponse);
            }
        });
    }

    /**
     * 单个人发送
     * @param userId
     * @param wsResponse
     */
    public void send(Integer userId, WsResponse wsResponse) {
        List<Integer> userIds = new ArrayList<Integer>(){
            {
                add(userId);
            }
        };
        MqMessage mqMessage = new MqMessage(userIds,wsResponse);
        this.broadcast(mqMessage);
    }

}